✨ Add WebSocket server to @effectionx/websocket#222
Conversation
Add useWebSocketServer(), the server counterpart of useWebSocket(). It yields a stream of incoming connections, each a full-duplex WebSocketResource, so client and server share the same handle type. The underlying server is supplied via a factory, keeping the package free of any concrete server dependency and platform-agnostic. BREAKING CHANGE: WebSocketResource.send() is now an Operation, invoked as `yield* socket.send(...)` on both client and server. Bumped to 3.0.0.
📝 WalkthroughWalkthroughAdds a WebSocket server stream API, updates ChangesWebSocket server and send() operation
Estimated code review effort: 3 (Moderate) | ~25 minutes 🚥 Pre-merge checks | ✅ 6✅ Passed checks (6 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
commit: |
| // client's `throw yield* once(socket, "error")` behavior | ||
| let errored = withResolvers<Error>(); | ||
| let onError = (error: Error) => errored.resolve(error); | ||
| server.on("error", onError); |
There was a problem hiding this comment.
we should be using once here from either @effectionx/node or effection depending on if it's TargetEvent or EventEmitter interface
There was a problem hiding this comment.
Done in 61f849d — now using once from @effectionx/node (the ws server is an EventEmitter): let [error] = yield* once<[Error]>(server, "error"); throw error;. Added @effectionx/node as a dependency.
| } | ||
| }); | ||
| }; | ||
| server.on("connection", onConnection); |
There was a problem hiding this comment.
we can use on here from effectionx/node or effection
There was a problem hiding this comment.
Done in 61f849d — connections now come from on(server, "connection") (from @effectionx/node), consumed with each.
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 4
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@websocket/server.test.ts`:
- Around line 117-120: The test is accessing `.data` on `firstMessages.next()`
and `secondMessages.next()` without narrowing the iterator result, but the
`.value` can also be a `CloseEvent`. Update the `received` setup in
`server.test.ts` to first unwrap each `IteratorResult` and assert/narrow that
the yielded value is a `MessageEvent<string>` before reading `.data`, using the
existing `firstMessages` and `secondMessages` iterators so the type checker can
distinguish them from `CloseEvent`.
In `@websocket/server.ts`:
- Around line 85-140: Add a test for connection-level error isolation in
useWebSocketServer: verify that a single accepted socket error after the
connection is established does not take down the server or other active
connections. Use the useWebSocketServer, onConnection, and the per-connection
useWebSocket<T> task behavior to assert the errored socket is isolated while the
server keeps accepting/serving remaining connections.
- Around line 108-119: The per-connection handling in onConnection currently
runs inside scope.run without its own failure boundary, so errors from
useWebSocket or the subscription drain loop can bubble up and affect the server.
Wrap the body of onConnection in a connection-local error boundary or
try/catch-style effect handler so each raw WebSocket failure is isolated, and
make sure the connection setup, connections.add(connection), and
subscription.next() loop all stay within that per-client scope.
In `@websocket/websocket.ts`:
- Around line 161-163: The public send method in websocket.ts is relying on
contextual typing for both its parameter and its Operation<void> return, so make
the types explicit on the send method in the websocket object literal. Update
the send signature in the websocket factory/adapter implementation to declare
the data parameter type and the void-returning operation type directly, keeping
the method compatible with the surrounding interface while following the
explicit public function typing guideline.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: ASSERTIVE
Plan: Pro
Run ID: 2406db94-4af3-4cb6-8f23-3452e452d764
📒 Files selected for processing (7)
websocket/README.mdwebsocket/mod.tswebsocket/package.jsonwebsocket/server.test.tswebsocket/server.tswebsocket/websocket.test.tswebsocket/websocket.ts
There was a problem hiding this comment.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
websocket/server.ts (1)
85-103: 🚀 Performance & Scalability | 🔵 TrivialUnbounded connection buffering could exhaust resources under connection floods.
createQueueis unlimited in size, and the accept loop wraps and stores every incoming socket viauseWebSocketregardless of whether anything ever reads fromserver.socket. If the consumer stalls or never iterates the stream, the server keeps accepting and holding live sockets (each with its own background tasks/listeners) indefinitely — a stalled or malicious client population could grow memory/FD usage without bound. Worth considering a cap on buffered connections (or backpressure signal) as a hardening follow-up.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@websocket/server.ts` around lines 85 - 103, The accept path in websocket/server.ts can buffer an unlimited number of live WebSocketResource entries, which risks unbounded memory and FD growth if server.socket is not being consumed. Update the createQueue/useWebSocket flow in the server accept loop to enforce a bounded connection buffer or apply backpressure so new sockets are not wrapped and retained indefinitely when the consumer stalls. Use the existing spawn, each(on(server, "connection")), and connections.add logic as the place to gate or reject excess connections.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Outside diff comments:
In `@websocket/server.ts`:
- Around line 85-103: The accept path in websocket/server.ts can buffer an
unlimited number of live WebSocketResource entries, which risks unbounded memory
and FD growth if server.socket is not being consumed. Update the
createQueue/useWebSocket flow in the server accept loop to enforce a bounded
connection buffer or apply backpressure so new sockets are not wrapped and
retained indefinitely when the consumer stalls. Use the existing spawn,
each(on(server, "connection")), and connections.add logic as the place to gate
or reject excess connections.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: ASSERTIVE
Plan: Pro
Run ID: e247a1c9-f01f-47c8-9884-ac73b73888d2
⛔ Files ignored due to path filters (1)
pnpm-lock.yamlis excluded by!**/pnpm-lock.yaml
📒 Files selected for processing (5)
websocket/package.jsonwebsocket/server.test.tswebsocket/server.tswebsocket/tsconfig.jsonwebsocket/websocket.ts
Phase 1: replace the unbounded wait for the peer close handshake in useWebSocket with timebox(), so a silent peer can no longer hang scope teardown. Adds @effectionx/timebox.
Phase 2: WebSocketResource gains a composable close(code?, reason?) operation; useWebSocketServer composes a 1001 "server shutting down" close for live connections on teardown. First-close-wins, so it takes precedence over the scope-exit 1000.
Phase 3: each accepted connection now runs inside a scoped() error boundary, so one socket erroring is contained instead of crashing the server. Failures are surfaced compositionally on a new server.errors stream rather than a callback.
There was a problem hiding this comment.
Actionable comments posted: 2
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
websocket/server.test.ts (1)
24-36: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueConsider extracting the repeated "accept a connection" boilerplate.
The
let { server, port } = yield* useTestServer(); let incoming = yield* server; let client = yield* connect(port); let connection = (yield* incoming.next()).value;sequence repeats across most tests in this file. A smallacceptConnection()helper (alongside the existinguseTestServer/connect/drainhelpers) would reduce duplication and keep new tests concise.Also applies to: 38-50, 101-113, 133-166
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@websocket/server.test.ts` around lines 24 - 36, The test setup in server.test.ts repeats the same “start server, wait for incoming, connect client, accept connection” boilerplate in multiple cases. Extract that sequence into a small helper such as acceptConnection() near the existing useTestServer, connect, and drain helpers, and update the affected tests to use it so the websocket tests stay concise and consistent.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@websocket/package.json`:
- Around line 17-19: The websocket package currently depends on the top-level
`@effectionx/node` entrypoint, which pulls in Node-only stream code through its
reexports and breaks browser/Deno consumers. Update the websocket package to
avoid importing the full `@effectionx/node` package and instead reference the
narrower `@effectionx/node/events` entrypoint, or separate the event helpers from
the Node stream adapter so `@effectionx/websocket` stays environment-agnostic.
In `@websocket/server.ts`:
- Around line 159-161: The teardown loop in the server shutdown path currently
closes each live connection sequentially via `connection.close(...)`, which can
stretch shutdown to N times the close timeout. Update the shutdown logic to run
the per-connection close effects concurrently using Effection’s `all` in the
`server.ts` teardown flow, and add `all` to the Effection import. Keep the
existing `close(1001, "server shutting down")` behavior inside the
per-connection task, and verify that the `close()` calls remain safe when
executed in parallel.
---
Outside diff comments:
In `@websocket/server.test.ts`:
- Around line 24-36: The test setup in server.test.ts repeats the same “start
server, wait for incoming, connect client, accept connection” boilerplate in
multiple cases. Extract that sequence into a small helper such as
acceptConnection() near the existing useTestServer, connect, and drain helpers,
and update the affected tests to use it so the websocket tests stay concise and
consistent.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: ASSERTIVE
Plan: Pro
Run ID: 8430e035-504d-4dd1-aea7-18e16862e950
⛔ Files ignored due to path filters (1)
pnpm-lock.yamlis excluded by!**/pnpm-lock.yaml
📒 Files selected for processing (6)
websocket/package.jsonwebsocket/server.test.tswebsocket/server.tswebsocket/tsconfig.jsonwebsocket/websocket.test.tswebsocket/websocket.ts
| "dependencies": { | ||
| "@effectionx/node": "workspace:*", | ||
| "@effectionx/timebox": "workspace:*" |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Check whether `@effectionx/node` itself depends on Node.js-specific runtime APIs
# or is purely a type/interface package usable in any JS environment.
fd package.json node --exec cat {}
rg -n "process\.|node:" node/*.ts node/mod.ts 2>/dev/nullRepository: thefrontside/effectionx
Length of output: 1604
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Inspect websocket package metadata and its source imports.
fd -a 'package.json' websocket
echo '--- websocket/package.json ---'
cat websocket/package.json
echo '--- websocket source imports ---'
rg -n 'from "`@effectionx/`(node|timebox)"|import .*`@effectionx/`(node|timebox)|require\\("`@effectionx/`(node|timebox)"\\)' websocketRepository: thefrontside/effectionx
Length of output: 1497
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo '--- websocket file map ---'
fd -a 'mod.ts|server.ts|client.ts|socket.ts|*.ts' websocket
echo '--- websocket/mod.ts ---'
cat -n websocket/mod.ts
echo '--- websocket/server.ts ---'
cat -n websocket/server.ts
echo '--- `@effectionx/timebox` package.json ---'
fd -a 'package.json' timebox
cat timebox/package.json
echo '--- `@effectionx/timebox` exports/imports ---'
rg -n 'export|import|setTimeout|clearTimeout|AbortController|node:' timebox
echo '--- `@effectionx/node` exports/imports ---'
cat -n node/mod.ts
rg -n 'export .*on|export .*once|import .*node:|from "node:' nodeRepository: thefrontside/effectionx
Length of output: 546
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo '--- websocket/mod.ts ---'
cat -n websocket/mod.ts
echo '--- websocket/server.ts ---'
cat -n websocket/server.ts
echo '--- websocket/websocket.ts ---'
cat -n websocket/websocket.ts
echo '--- timebox/package.json ---'
cat timebox/package.json
echo '--- timebox/mod.ts ---'
cat -n timebox/mod.tsRepository: thefrontside/effectionx
Length of output: 20994
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo '--- node/mod.ts ---'
cat -n node/mod.ts
echo '--- node/events.ts ---'
cat -n node/events.ts
echo '--- node/stream.ts ---'
cat -n node/stream.tsRepository: thefrontside/effectionx
Length of output: 9186
Avoid a hard dependency on @effectionx/node in websocket/package.json. Its top-level entrypoint reexports node/stream.ts, so importing @effectionx/websocket pulls in node:stream even for browser/Deno consumers. Import @effectionx/node/events instead, or split the event helpers from the Node-only stream adapter.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@websocket/package.json` around lines 17 - 19, The websocket package currently
depends on the top-level `@effectionx/node` entrypoint, which pulls in Node-only
stream code through its reexports and breaks browser/Deno consumers. Update the
websocket package to avoid importing the full `@effectionx/node` package and
instead reference the narrower `@effectionx/node/events` entrypoint, or separate
the event helpers from the Node stream adapter so `@effectionx/websocket` stays
environment-agnostic.
| for (let connection of [...live]) { | ||
| yield* connection.close(1001, "server shutting down"); | ||
| } |
There was a problem hiding this comment.
🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick win
Close live connections concurrently during teardown to bound shutdown latency.
Each yield* connection.close(1001, ...) awaits that connection's close handshake, bounded by CLOSE_TIMEOUT_MS. Done sequentially, a server with N connections whose peers go silent takes up to N * CLOSE_TIMEOUT_MS to release, delaying scope exit. Running them concurrently with all bounds worst-case teardown to a single timeout.
♻️ Proposed fix
- for (let connection of [...live]) {
- yield* connection.close(1001, "server shutting down");
- }
+ yield* all(
+ [...live].map((connection) =>
+ connection.close(1001, "server shutting down")
+ ),
+ );Add all to the effection import:
import {
+ all,
createQueue,
createSignal,
each,
resource,
scoped,
spawn,
} from "effection";Please confirm all is exported by effection 4.0.0 and that concurrent close() calls don't contend on shared teardown state.
📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| for (let connection of [...live]) { | |
| yield* connection.close(1001, "server shutting down"); | |
| } | |
| yield* all( | |
| [...live].map((connection) => | |
| connection.close(1001, "server shutting down") | |
| ), | |
| ); |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@websocket/server.ts` around lines 159 - 161, The teardown loop in the server
shutdown path currently closes each live connection sequentially via
`connection.close(...)`, which can stretch shutdown to N times the close
timeout. Update the shutdown logic to run the per-connection close effects
concurrently using Effection’s `all` in the `server.ts` teardown flow, and add
`all` to the Effection import. Keep the existing `close(1001, "server shutting
down")` behavior inside the per-connection task, and verify that the `close()`
calls remain safe when executed in parallel.
There was a problem hiding this comment.
In general this is a good idea, but I believe that .close() is a synchronous operation.
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
fd server.ts websocket -x cat -n {}
echo "-----websocket.ts-----"
fd websocket.ts websocket -x cat -n {}Length of output: 16955
🏁 Script executed:
#!/bin/bash
fd package.json websocket -x cat {}
echo "---"
rg -n "effection" websocket/package.json
echo "--- check effection exports for 'all' ---"
find / -path /proc -prune -o -type d -name "effection" -print 2>/dev/null | head -20Length of output: 1783
🤖 web_search executed:
Does Effection (effection npm package) version 3 and 4 export an all function for running multiple operations concurrently?
Length of output: 7341
Confirmed all is exported by Effection in both v3 and v4 (it's been a core API since v3, with a fix for its typing in 3.0.1), so that part of my suggestion is safe.
On the substance of your point: looking at websocket.ts, WebSocketResource.close() (Lines 191-193) is a generator method that calls closeSocket(), which does socket.close(code, reason) (the synchronous, fire-and-forget native call) and then yield* timebox(CLOSE_TIMEOUT_MS, () => closed) (Lines 158-160) — waiting up to CLOSE_TIMEOUT_MS (1000ms) for the peer's close handshake to resolve. So while the underlying socket.close() call is synchronous, the WebSocketResource.close() operation itself is not — it suspends until the close event fires or the timebox expires.
That's exactly why sequential yield* connection.close(...) in server.ts's teardown loop (Lines 159-161) can cost up to N * CLOSE_TIMEOUT_MS if there are N live connections with unresponsive peers, since each iteration blocks on its own timebox before starting the next. Running them concurrently with all would bound that to a single CLOSE_TIMEOUT_MS regardless of N.
Given close() calls only touch per-connection state (each connection's own socket and closed resolver), concurrent execution shouldn't contend on shared teardown state — live is only read (via the [...live] snapshot) before the closes are kicked off, not mutated by close() itself.
| * | ||
| * await main(function* () { | ||
| * let server = yield* useWebSocketServer<string>( | ||
| * () => new WebSocketServer({ port: 3000 }) as unknown as WebSocketServerLike, |
There was a problem hiding this comment.
Why do we need to cast here? Is WebSocketServer not WebSocketServerLike?
| }); | ||
|
|
||
| // accept connections. Each is handled in its own task wrapped in `scoped`, | ||
| // which is a real error boundary (its trap/delimiter contains a crash) — so |
There was a problem hiding this comment.
trap/delimiter are private concepts and should not be referenced in comments
| for (let connection of [...live]) { | ||
| yield* connection.close(1001, "server shutting down"); | ||
| } |
| for (let connection of [...live]) { | ||
| yield* connection.close(1001, "server shutting down"); | ||
| } |
There was a problem hiding this comment.
In general this is a good idea, but I believe that .close() is a synchronous operation.
| yield* provide({ | ||
| *[Symbol.iterator]() { | ||
| return connections; | ||
| }, | ||
| errors, |
There was a problem hiding this comment.
The problem with this is that it subtly breaks the stream contract which is that it tends to be stateless, and that when you subscribe to a stream, that is when state is activated.
Is it not better to make the web socket server a subscription, which is more honest to the consumer that "this will allocate state and begin filling a buffer"?
| live.add(connection); | ||
| connections.add(connection); |
There was a problem hiding this comment.
Why are we storing these connections in two separate sets?
2 participants
Motivation
The
@effectionx/websocketpackage shipped only a client (useWebSocket()). There was no server counterpart, even though the package's own test file already implemented the full server-side mechanic by hand — awsWebSocketServerfiringconnectionevents, each raw socket wrapped withuseWebSocket(). This PR formalizes that pattern into a reusable server that pairs naturally with the existing client.Approach
New
useWebSocketServer()(websocket/server.ts) — the server counterpart ofuseWebSocket():WebSocketServerResource<T>: an EffectionStreamwhose items are the same full-duplexWebSocketResourcethe client produces, so both sides share one handle type (iterate to receive,yield* connection.send()to reply).useWebSocket()to wrap each incoming socket.createQueue, so none are dropped between the server starting to listen and the consumer iterating.error, mirroring the client.() => WebSocketServerLikefactory, so the package never imports a concrete server implementation and stays platform-agnostic — no new runtime/peer dependency (wsremains a devDependency for tests only).sendis now anOperation— invoked asyield* resource.send(...)on both client and server, keeping the sharedWebSocketResourcesymmetric and letting sends participate in structured concurrency.Warning
Breaking change:
WebSocketResource.send()changed from a synchronousvoidcall to anOperation. Callers must now writeyield* socket.send(...). Version bumped2.3.3 → 3.0.0.Also included:
mod.tsre-exports the server; updated client README/tests toyield*their sends.server.test.ts(5 tests): connection delivery, message round-trips in both directions, close-propagation on client disconnect, server teardown closing live clients, and multiple simultaneous clients as distinct connections.Verification
wsserver against a real nativeWebSocketclient, exercising the pairing end-to-end.biome lintclean,biome formatclean,tsc -bpasses.Summary by CodeRabbit
WebSocketResource.sendto return an operation, and added an explicitWebSocketResource.close(code?, reason?).